package br.ufc.great.dtx.tm.sch;

import java.io.IOException;
import java.io.ObjectOutputStream;
import java.net.Socket;
import java.util.Vector;

import br.ufc.great.Graph;
import br.ufc.great.dtx.server.DTXOutput;
import br.ufc.great.dtx.tm.AbortTransactionException;
import br.ufc.great.dtx.tm.lm.LockIncompatibleException;
import br.ufc.great.dtx.tm.lm.LockManager;
import br.ufc.great.dtx.tm.operations.Operation;
import br.ufc.great.dtx.tm.operations.Site;

/**
 * 
 * @author Leonardo Oliveira Moreira
 *
 * Classe que implementa uma parte do escalonador que processa as operações remotas
 */
public class SchedulerRemoteOperation extends Thread {

	private DTXOutput output;
	private Scheduler scheduler;
	private LockManager lockManager;
	
	// Lista de operações remotas
	private Vector<RemoteOperation> remoteOperations;
	
	public SchedulerRemoteOperation(DTXOutput output, Scheduler scheduler, LockManager lockManager) {
		this.output = output;
		this.scheduler = scheduler;
		this.lockManager = lockManager;
		remoteOperations = new Vector<RemoteOperation>();
		setPriority(MIN_PRIORITY);
	}
	
	public void run() {
		while (isAlive()) {
			// Processa operações remotas
			while (remoteOperations.size() > 0) {
				int j = 0;
				RemoteOperation remoteOperation = null;
				synchronized (remoteOperations) {
					for (j = 0; j < remoteOperations.size(); j++) {
						RemoteOperation temp = remoteOperations.elementAt(j);
						if (temp.getOperation().getState() == Operation.STATE_NONE) {
							remoteOperation = temp;
							break;
						}
					}
				}
				if (remoteOperation == null) {
					pause();
					break;
				}
				try {
					processRemoteOperation(remoteOperation.getOperation());
				}
				catch (AbortTransactionException e) {
					synchronized (lockManager) {
						lockManager.unlock(remoteOperation.getOperation().getTransactionId(), remoteOperation.getOperation().getId());
					}
					remoteOperation.getOperation().setState(Operation.STATE_ABORTED);
				}
				catch (LockIncompatibleException e) {
					synchronized (lockManager) {
						lockManager.unlock(remoteOperation.getOperation().getTransactionId(), remoteOperation.getOperation().getId());
					}
					Graph waitFor = scheduler.getWaitForGraph();
					synchronized (waitFor) {
						waitFor.addLink(remoteOperation.getOperation().getTransactionId(), e.getTransactionIdIncompatible());
					}
				}
				output.print(getClass().getSimpleName(), "Remote operation " + remoteOperation.getOperation().toString() + " " + "finished", DTXOutput.TEST);
				Socket socket = remoteOperation.getSocket();
				try {
					Operation operation = remoteOperation.getOperation();
					Vector<Site> sites = operation.getSites();
					operation.setSites(new Vector<Site>());
					ObjectOutputStream out = new ObjectOutputStream(socket.getOutputStream());
					out.writeObject(remoteOperation.getOperation());
					out.flush();
					synchronized (remoteOperations) {
						remoteOperations.remove(remoteOperation);						
					}
					operation.setSites(sites);
				} catch (IOException e) {
					e.printStackTrace();
					output.print(getClass().getSimpleName(), e.getMessage(), DTXOutput.ERROR);
				}
			}
			if (remoteOperations.size() > 0)
				continue;
			pause();
		}
	}
	
	public synchronized void pause() {
		try {
			output.print(getClass().getSimpleName(), "is waiting", DTXOutput.TEST);
			wait();
		}
		catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
	
	public synchronized void restart() {
		notify();
	}
	
	public synchronized void addRemoteOperation(RemoteOperation remoteOperation) {
		Operation operation = remoteOperation.getOperation();
		output.print(getClass().getSimpleName(), "Remote operation " + operation.getId() + " of transaction " + operation.getTransactionId() + " was received", DTXOutput.TEST);
		remoteOperations.add(remoteOperation);
		restart();
	}

	
	private void processRemoteOperation(Operation operation) throws AbortTransactionException, LockIncompatibleException {
		String result = null;
		try {
			output.print(getClass().getSimpleName(), "Remote operation " + operation.toString() + " " + "is processing", DTXOutput.TEST);
			synchronized (lockManager) {
				output.print(getClass().getSimpleName(), "Enter in lockManager " + operation.toString() + " " + "for this processing", DTXOutput.TEST);
				result = lockManager.processOperation(operation);
				output.print(getClass().getSimpleName(), "Exit in lockManager " + operation.toString() + " ", DTXOutput.TEST);
			}
			output.print(getClass().getSimpleName(), "Remote operation " + operation.toString() + " " + "executed", DTXOutput.TEST);
			operation.setResult(result);
			operation.setState(Operation.STATE_EXECUTED);
		}
		catch (LockIncompatibleException e) {
			output.print(getClass().getSimpleName(), "Lock incompatible for " + operation.toString() + " ", DTXOutput.TEST);
			throw e;
		}
	}
	
	public synchronized int size() {
		return remoteOperations.size();
	}
	
}
